For more robust detection of awaitable results from _compute_watch_status, it's better to use inspect.isawaitable() instead of asyncio.iscoroutine(). isawaitable() is more general and correctly handles not just coroutines from async def functions, but also other awaitable objects like asyncio.Future or custom objects with an __await__ method. This makes the mixin more resilient to different implementation patterns in subclasses.

You'll need to add import inspect at the top of the file.

Accepted. Switched to inspect.isawaitable() -- it's the idiomatic pattern for this and handles the sync/async dispatch more robustly.

Note: the broader rationale about asyncio.Future and custom __await__ objects doesn't apply here since _compute_watch_status only ever returns int (sync) or a coroutine (async def), but isawaitable is still the right call.

Catch asyncio.TimeoutError in Watch poll loop

Watch() currently catches built-in TimeoutError, but on Python 3.10 (which is supported via requires-python >=3.10) asyncio.wait_for() raises asyncio.TimeoutError instead. When a stream is healthy and no shutdown event occurs for one poll interval, that timeout escapes the loop and aborts the RPC rather than continuing to poll, so long-lived watch streams terminate unexpectedly in normal operation.

Useful? React with 👍 / 👎.

Good catch -- this is a real bug on Python 3.10.

On 3.10, asyncio.TimeoutError inherits from Exception directly, NOT from the builtin TimeoutError. So except TimeoutError: misses it entirely, causing the poll loop to crash on every interval when no shutdown event fires. On 3.11+, they unified the two (asyncio.TimeoutError is builtins.TimeoutError), which is why this wasn't caught in local testing.

Fixed: except asyncio.TimeoutError: -- works on both 3.10 and 3.11+ since it's the same class on 3.11+.

Emit shutdown status before terminating Watch stream

This early return can drop a real status transition during shutdown: if set_not_serving() runs after status was computed for the current iteration but before this check executes, the stream exits immediately without sending the final NOT_SERVING update. In that race window, Watch clients only observe EOF and miss the health-state change event they rely on for routing decisions.

Useful? React with 👍 / 👎.

Evaluated -- pushing back on this one.

The race requires set_not_serving() to fire between _resolve_watch_status() returning and _is_shutting_down() executing. In asyncio's single-threaded cooperative model:

• SGLang: _compute_watch_status() is sync -- no yield point, race is impossible.
• vLLM: async via await check_health(), but the window between the return of _resolve_watch_status() and the _is_shutting_down() check is a single Python statement with no await -- no coroutine switch can happen there.

Even in the theoretical case where shutdown lands between _resolve_watch_status yielding control (during check_health()) and the shutdown check: the client would see SERVING then EOF. Any conformant Watch client treats EOF as "server unavailable" -- the gRPC transport-level disconnect is the primary shutdown signal, not an in-band NOT_SERVING message.

Not worth the added complexity.

🧹 Nitpick | 🔵 Trivial

Bound stream-consumption waits to prevent indefinite test hangs on regressions.

These async for loops are open-ended and rely on side effects to terminate. If a future change breaks shutdown signaling, this module can hang instead of failing fast. Wrap stream collection in asyncio.wait_for(...) (or a shared helper) with a small timeout.

+async def _collect_watch(servicer, request_msg, grpc_context, timeout_s=1.0):
+    async def _run():
+        items = []
+        async for response in servicer.Watch(request_msg, grpc_context):
+            items.append(response.status)
+        return items
+    return await asyncio.wait_for(_run(), timeout=timeout_s)

Also applies to: 63-66, 83-86, 120-123, 144-148, 172-175

Verify each finding against the current code and only fix it if needed.

In `@grpc_servicer/tests/test_sglang_health_watch.py` around lines 41 - 45, The
test uses open-ended async for loops over servicer.Watch which can hang if
shutdown signaling breaks; replace these with a bounded collector using
asyncio.wait_for by creating a helper (e.g., _collect_watch) that runs the async
iterator and accumulates response.status, then call
asyncio.wait_for(_collect_watch(servicer.Watch(request_msg, grpc_context)),
timeout=SHORT_TIMEOUT) in tests; update occurrences referencing servicer.Watch
and servicer.set_not_serving (lines around the current loops) to use the helper
so tests fail fast instead of hanging.

🧹 Nitpick | 🔵 Trivial

Add explicit timeout bounds around stream collection to avoid stuck CI jobs.

These tests currently depend on Watch termination side effects; if termination regresses, they can hang. Use asyncio.wait_for (directly or via helper) so failures surface deterministically.

Also applies to: 68-71, 91-94, 126-129, 147-151, 162-166

Verify each finding against the current code and only fix it if needed.

In `@grpc_servicer/tests/test_vllm_health_watch.py` around lines 34 - 39, The
streaming test uses servicer.Watch and can hang if the stream never terminates;
wrap the async collection in an explicit timeout using asyncio.wait_for (or your
existing async helper) so the test fails fast. Specifically, change the block
that iterates over async for response in servicer.Watch(request_msg,
grpc_context) (and similar blocks around servicer.set_not_serving) to collect
the responses inside an await asyncio.wait_for(..., timeout=some_seconds) call
(e.g., gather into a coroutine that consumes one response then calls
servicer.set_not_serving) so the test raises on timeout instead of hanging.